package com.cpr.demo.mycraft.mq.redis.single;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.cpr.demo.constant.MyMqConstant;
import com.cpr.demo.constant.RedisKeyConstant;
import com.cpr.demo.mycraft.mq.redis.MessageHandler;
import com.cpr.demo.mycraft.mq.redis.RedisMessage;
import com.cpr.demo.mycraft.mq.redis.RedisMessageQueueManager;
import com.cpr.demo.service.mq.MqReportService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Chen Panrui
 * created at 2021/10/9 13:52
 * Q: 为什么会有这个类？
 * A: 这是一个基于 redis.list实现的(pub/sub)，一个message只会被消费一次的MQ
 *
 * 测试demo-订阅某个queue {@link com.cpr.demo.service.PersonCacheServiceImpl#afterPropertiesSet()}
 * 测试demo-发消息给某个queue {@link com.cpr.demo.controller.MqTestController#send1(String, String)}
 * 测试数据保存-日志表 {@link com.cpr.demo.model.mq.MqReport}
 */
@Slf4j
@RequiredArgsConstructor
@Component("singleRedisMQ")
public class SingleRedisMessageQueueManagerImpl implements RedisMessageQueueManager, InitializingBean, DisposableBean {

    private final StringRedisTemplate stringRedisTemplate;
    private final AtomicBoolean isRunning = new AtomicBoolean(true);
    private ExecutorService subscribeThreadPool;
    /**
     * 每个应用 对应 一个实例
     * 我需要知道，是哪个机器 消费了这个message，serverId用于记录
     */
    private final String serverId;
    private final Long mqMaxHold;
    private final MqReportService mqReportService;

    @Override
    public void publish(String queue, RedisMessage message) {
        // 记录数据
        mqReportService.save(RedisKeyConstant.MSG_SINGLE_QUEUE + queue, message, serverId, MyMqConstant.MQ_ROLE_SENDER, MyMqConstant.MQ_TYPE_REDIS_SINGLE);
        log.info("应用:{},发布消息到队列{}", serverId, RedisKeyConstant.MSG_SINGLE_QUEUE + queue);
        stringRedisTemplate.boundListOps(RedisKeyConstant.MSG_SINGLE_QUEUE + queue).leftPush(JSONUtil.toJsonStr(message));
        stringRedisTemplate.expire(RedisKeyConstant.MSG_SINGLE_QUEUE + queue, MyMqConstant.EXPIRE_TIME_60, TimeUnit.MINUTES);
    }

    @Override
    public void subscribe(String queue, MessageHandler handler) {
        log.info("应用:{},订阅redis消息队列{}...", serverId, RedisKeyConstant.MSG_SINGLE_QUEUE + queue);
        /*
         * 你不能使用当前线程，必须单独开一个线程来执行
         */
        subscribeThreadPool.execute(()->{
            while (isRunning.get()){
                try {
                    String jsonMessage = stringRedisTemplate.boundListOps(RedisKeyConstant.MSG_SINGLE_QUEUE + queue).rightPop(mqMaxHold, TimeUnit.SECONDS);
                    if(StrUtil.isNotBlank(jsonMessage)){
                        // 记录数据
                        mqReportService.save(RedisKeyConstant.MSG_SINGLE_QUEUE + queue, JSONUtil.toBean(jsonMessage, RedisMessage.class), serverId, MyMqConstant.MQ_ROLE_RECEIVE, MyMqConstant.MQ_TYPE_REDIS_SINGLE);
                        log.info("应用:{},处理{}队列获取message={}", serverId, RedisKeyConstant.MSG_SINGLE_QUEUE + queue, jsonMessage);
                        handler.handle(jsonMessage);
                    }
                } catch (Exception e){
                    // TODO 优化点：抛异常可以塞到redis里面，重新在处理一次
                    log.error("应用:{},监听队列{}异常", serverId, RedisKeyConstant.MSG_SINGLE_QUEUE + queue , e);
                }
            }
            log.info("应用:{},停止订阅队列{}", serverId, RedisKeyConstant.MSG_SINGLE_QUEUE + queue);
        });
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.subscribeThreadPool = new ThreadPoolExecutor(1, Integer.MAX_VALUE,
                1L, TimeUnit.HOURS,
                new SynchronousQueue<>(),
                new ThreadFactoryBuilder().setNameFormat("single-consume-redis-MQ").build(),
                new ThreadPoolExecutor.AbortPolicy());
    }

    @Override
    public void destroy() throws Exception {
        // TODO 优化点：服务中断，重启 需要保存正在消费&&没有消费完成的消息，重启的时候使用新线程处理
        log.info(">>>destroy单消费的redis-MQ");
        this.isRunning.compareAndSet(true, false);
        subscribeThreadPool.shutdown();
    }
}
